#include "kafka.hpp"
#include <iostream>
#include <thread>
#include <chrono>
#include <librdkafka/rdkafkacpp.h>
#include <algorithm>
using namespace std;

Kafka::Kafka() : producer_(nullptr), consumer_(nullptr), producer_conf_(nullptr), consumer_conf_(nullptr), running_(false)
{
}

Kafka::~Kafka(){
    if (running_) {
        stop();
    }
    
    // 清理生产者
    if (producer_ != nullptr) {
        delete producer_;
        producer_ = nullptr;
    }
    
    // 清理消费者
    if (consumer_ != nullptr) {
        delete consumer_;
        consumer_ = nullptr;
    }
    
    // 清理配置
    if (producer_conf_ != nullptr) {
        delete producer_conf_;
        producer_conf_ = nullptr;
    }
    
    if (consumer_conf_ != nullptr) {
        delete consumer_conf_;
        consumer_conf_ = nullptr;
    }
}

bool Kafka::init(const string& brokers) {
    brokers_ = brokers;
    
    // 初始化生产者配置
    producer_conf_ = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    string errstr;
    
    if (producer_conf_->set("bootstrap.servers", brokers_, errstr) != RdKafka::Conf::CONF_OK) {
        cerr << "Producer conf set bootstrap.servers failed: " << errstr << endl;
        return false;
    }
    
    // 创建生产者实例
    producer_ = RdKafka::Producer::create(producer_conf_, errstr);
    if (producer_ == nullptr) {
        cerr << "Failed to create producer: " << errstr << endl;
        return false;
    }
    
    cout << "Kafka producer created successfully" << endl;
    
    // 初始化消费者配置
    consumer_conf_ = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    
    if (consumer_conf_->set("bootstrap.servers", brokers_, errstr) != RdKafka::Conf::CONF_OK) {
        cerr << "Consumer conf set bootstrap.servers failed: " << errstr << endl;
        return false;
    }
    
    if (consumer_conf_->set("group.id", "chat_service_group", errstr) != RdKafka::Conf::CONF_OK) {
        cerr << "Consumer conf set group.id failed: " << errstr << endl;
        return false;
    }
    
    if (consumer_conf_->set("enable.auto.commit", "true", errstr) != RdKafka::Conf::CONF_OK) {
        cerr << "Consumer conf set enable.auto.commit failed: " << errstr << endl;
        return false;
    }
    
    if (consumer_conf_->set("auto.offset.reset", "latest", errstr) != RdKafka::Conf::CONF_OK) {
        cerr << "Consumer conf set auto.offset.reset failed: " << errstr << endl;
        return false;
    }
    
    // 创建消费者实例
    consumer_ = RdKafka::KafkaConsumer::create(consumer_conf_, errstr);
    if (consumer_ == nullptr) {
        cerr << "Failed to create consumer: " << errstr << endl;
        return false;
    }
    
    cout << "Kafka consumer created successfully" << endl;
    this->start_consuming();//开始消费
    return true;
}

bool Kafka::publish(const string& topic, int channel, const string& message) {
    if (producer_ == nullptr) {
        cerr << "Kafka producer not initialized" << endl;
        return false;
    }
    
    // 检查topic缓存
    {
        lock_guard<mutex> lock(producer_mutex_);
        if (producer_topic_cache_.find(topic) == producer_topic_cache_.end()) {
            string errstr;
            RdKafka::Topic* rdk_topic = RdKafka::Topic::create(producer_, topic, nullptr, errstr);
            if (rdk_topic == nullptr) {
                cerr << "Failed to create topic " << topic << ": " << errstr << endl;
                return false;
            }
            producer_topic_cache_[topic] = unique_ptr<RdKafka::Topic>(rdk_topic);
        }
    }
    
    // 构造带channel的消息
    string channel_message = to_string(channel) + ":" + message;
    
    // 发送消息
    RdKafka::ErrorCode err = producer_->produce(
        topic,
        RdKafka::Topic::PARTITION_UA,  // 使用默认分区
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char*>(channel_message.c_str()),
        channel_message.size(),
        nullptr,  // key
        0,        // key length
        0,        // timestamp
        nullptr   // msg_opaque
    );
    
    if (err != RdKafka::ERR_NO_ERROR) {
        cerr << "Failed to produce message to topic " << topic << ": " << RdKafka::err2str(err) << endl;
        return false;
    }
    
    // 等待消息发送完成
    producer_->flush(1000);
    return true;
}

bool Kafka::subscribe(const string& topic) {
    if (consumer_ == nullptr) {
        cerr << "Kafka consumer not initialized" << endl;
        return false;
    }
    
    // 添加到订阅列表
    {
        lock_guard<mutex> lock(consumer_mutex_);
        if (find(subscribed_topics_.begin(), subscribed_topics_.end(), topic) == subscribed_topics_.end()) {
            subscribed_topics_.push_back(topic);
        }
    }
    
    // 让消费者订阅主题
    vector<string> topics;
    {
        lock_guard<mutex> lock(consumer_mutex_);
        topics = subscribed_topics_;
    }
    
    RdKafka::ErrorCode err = consumer_->subscribe(topics);
    if (err != RdKafka::ERR_NO_ERROR) {
        cerr << "Failed to subscribe to topic " << topic << ": " << RdKafka::err2str(err) << endl;
        return false;
    }
    
    cout << "Subscribed to topic: " << topic << endl;
    return true;
}

bool Kafka::unsubscribe(const string& topic) {
    if (consumer_ == nullptr) {
        cerr << "Kafka consumer not initialized" << endl;
        return false;
    }
    
    // 从订阅列表中移除
    {
        lock_guard<mutex> lock(consumer_mutex_);
        auto it = find(subscribed_topics_.begin(), subscribed_topics_.end(), topic);
        if (it != subscribed_topics_.end()) {
            subscribed_topics_.erase(it);
        }
    }
    
    // 重新订阅剩余的主题
    vector<string> topics;
    {
        lock_guard<mutex> lock(consumer_mutex_);
        topics = subscribed_topics_;
    }
    
    RdKafka::ErrorCode err = consumer_->subscribe(topics);
    if (err != RdKafka::ERR_NO_ERROR) {
        cerr << "Failed to update subscription: " << RdKafka::err2str(err) << endl;
        return false;
    }
    
    cout << "Unsubscribed from topic: " << topic << endl;
    return true;
}

void Kafka::start_consuming() {
    if (consumer_ == nullptr || notify_handler_ == nullptr) {
        cerr << "Consumer not initialized or notify handler not set" << endl;
        return;
    }
    
    running_ = true;
    
    // 启动独立线程接收消息
    consuming_thread_ = thread([this]() {
        observe_topic_messages();
    });
    
    cout << "Kafka consuming thread started" << endl;
}

void Kafka::stop() {
    running_ = false;
    
    if (consumer_ != nullptr) {
        consumer_->close();
    }
    
    if (consuming_thread_.joinable()) {
        consuming_thread_.join();
    }
    
    cout << "Kafka consuming stopped" << endl;
}

void Kafka::observe_topic_messages() {
    if (consumer_ == nullptr) {
        cerr << "Kafka consumer not initialized" << endl;
        return;
    }
    
    cout << "Starting to observe Kafka messages..." << endl;
    
    while (running_) {
        RdKafka::Message* msg = consumer_->consume(1000); // 1秒超时
        
        switch (msg->err()) {
            case RdKafka::ERR_NO_ERROR: {
                // 处理收到的消息
                string payload(static_cast<const char*>(msg->payload()), msg->len());
                
                // 解析消息格式：channel:message
                size_t colon_pos = payload.find(':');
                if (colon_pos != string::npos) {
                    int channel = stoi(payload.substr(0, colon_pos));
                    string message = payload.substr(colon_pos + 1);
                    notify_handler_(channel, message);
                } else {
                    // 如果没有找到channel分隔符，将整个消息作为内容处理
                    notify_handler_(0, payload);
                }
                break;
            }
            case RdKafka::ERR__TIMED_OUT:
                // 超时，继续循环
                break;
            case RdKafka::ERR__PARTITION_EOF:
                // 分区结束，继续循环
                break;
            default:
                cerr << "Kafka consume error: " << msg->errstr() << endl;
                break;
        }
        
        delete msg;
    }
    
    cout << "Kafka observe thread exiting" << endl;
}

void Kafka::init_notify_handler(kafka_handler handler) {
    notify_handler_ = handler;
}